set hive.exec.dynamic.partition = true;
set hive.exec.dynamic.partition.mode = 'nonstrict';
set hive.exec.max.dynamic.partitions = 400;
set hive.exec.max.dynamic.partitions.pernode = 200;

with warehouse_network_mid as (
    select final_sign_network_code
         , final_sign_network_name
         , end_agent_code
         , end_agent_name
         , need_sign_count_1d
         , retention_count_1d
         , need_sign_count_2d
         , retention_count_2d
         , need_sign_count_3d
         , retention_count_3d
         , need_sign_count_5d
         , retention_count_5d
         , need_sign_count_7d
         , retention_count_7d
         , scan_sum
         , staff_code
         , staff_name
         , position_name
         , is_warn
         , final_plan_sign_date
         , agent_type
         , virt_code
         , virt_name
         , end_provider_id
         , end_provider_name
         , end_city_id
         , end_city_name
         , district_code
         , district_desc
         , district_id
         , area_code
         , area_name
    from jms_dm.dm_mng_explosion_warehouse_network_dt
    where dt = '{{ execution_date | cst_ds }}'
),
     warehouse_network_punish as (
         select final_sign_network_code
         from jms_dm.dm_mng_explosion_warehouse_network_punish_dt
         where dt = '{{ execution_date | cst_ds }}'
         group by final_sign_network_code
     )
insert
overwrite
table
jms_dm.dm_mng_explosion_warehouse_network_second_dt
partition
(
dt
)
select t1.final_sign_network_code
     , t1.final_sign_network_name
     , t1.end_agent_code
     , t1.end_agent_name
     , t1.need_sign_count_1d
     , t1.retention_count_1d
     , t1.need_sign_count_2d
     , t1.retention_count_2d
     , t1.need_sign_count_3d
     , t1.retention_count_3d
     , t1.need_sign_count_5d
     , t1.retention_count_5d
     , t1.need_sign_count_7d
     , t1.retention_count_7d
     , t1.scan_sum
     , t1.staff_code
     , t1.staff_name
     , t1.position_name
     , t1.is_warn
     , t1.final_plan_sign_date
     , t1.agent_type
     , t1.virt_code
     , t1.virt_name
     , t1.end_provider_id
     , t1.end_provider_name
     , t1.end_city_id
     , t1.end_city_name
     , t1.district_code
     , t1.district_desc
     , t1.district_id
     , null                            as duration
     , t1.area_code
     , t1.area_name
     , '{{ execution_date | cst_ds }}' as dt
from warehouse_network_mid t1
         left anti
         join warehouse_network_punish t2
              on t1.final_sign_network_code = t2.final_sign_network_code
;

insert
    overwrite
    table jms_dm.dm_mng_explosion_warehouse_network_second_dt
    partition
    (dt)
select t1.final_sign_network_code
     , t1.final_sign_network_name
     , t1.end_agent_code
     , t1.end_agent_name
     , t1.need_sign_count_1d
     , t1.retention_count_1d
     , t1.need_sign_count_2d
     , t1.retention_count_2d
     , t1.need_sign_count_3d
     , t1.retention_count_3d
     , t1.need_sign_count_5d
     , t1.retention_count_5d
     , t1.need_sign_count_7d
     , t1.retention_count_7d
     , t1.scan_sum
     , t1.staff_code
     , t1.staff_name
     , t1.position_name
     , t1.is_warn
     , t1.final_plan_sign_date
     , t1.agent_type
     , t1.virt_code
     , t1.virt_name
     , t1.end_provider_id
     , t1.end_provider_name
     , t1.end_city_id
     , t1.end_city_name
     , t1.district_code
     , t1.district_desc
     , t1.district_id
     , t2.duration
     , t1.area_code
     , t1.area_name
     , t1.dt
from (
         select *
         from jms_dm.dm_mng_explosion_warehouse_network_second_dt
         where dt = '{{ execution_date | cst_ds }}'
     ) t1
         left join
     (
         select final_sign_network_code
              , max(duration)     as duration
              , max(maxdate_time) as max_date_time
         from (
                  select final_sign_network_code
                       , date_sub(final_plan_sign_date, rn) as diff_date
                       , count(1)                           as duration
                       , max(final_plan_sign_date)          as maxdate_time
--                        , min(final_plan_sign_date)          as mindate_time
                  from (
                           select final_sign_network_code
                                , final_plan_sign_date
                                , row_number()
                                   over (partition by final_sign_network_code order by final_plan_sign_date asc) as rn
                           from jms_dm.dm_mng_explosion_warehouse_network_second_dt
                           where dt >= date_add('{{ execution_date | cst_ds }}', -61)
                             and dt <= '{{ execution_date | cst_ds }}'
                       ) t
                  group by final_sign_network_code, diff_date
              ) t
         where maxdate_time = '{{ execution_date | cst_ds }}'
         group by final_sign_network_code) t2
     on t1.final_sign_network_code = t2.final_sign_network_code
    distribute by dt
;

--汇总
with day as (
    select final_sign_network_code,
           end_agent_code,
           end_agent_name,
           nvl(virt_code, end_agent_code) as virt_code,
           nvl(virt_name, end_agent_name) as virt_name,
           end_provider_id,
           end_provider_name,
           district_code,
           district_desc,
           area_code,
           area_name
    from jms_dm.dm_mng_explosion_warehouse_network_second_dt
    where dt = '{{ execution_date | cst_ds }}'
    group by final_sign_network_code,
             end_agent_code,
             end_agent_name,
             virt_code,
             virt_name,
             end_provider_id,
             end_provider_name,
             district_code,
             district_desc,
             area_code,
             area_name
),
     last_day as (
         select final_sign_network_code,
                end_agent_code,
                end_agent_name,
                nvl(virt_code, end_agent_code) as virt_code,
                nvl(virt_name, end_agent_name) as virt_name,
                end_provider_id,
                end_provider_name,
                district_code,
                district_desc,
                area_code,
                area_name
         from jms_dm.dm_mng_explosion_warehouse_network_second_dt
         where dt = date_add('{{ execution_date | cst_ds }}', -1)
         group by final_sign_network_code,
                  end_agent_code,
                  end_agent_name,
                  virt_code,
                  virt_name,
                  end_provider_id,
                  end_provider_name,
                  district_code,
                  district_desc,
                  area_code,
                  area_name
     ),
     union_all_tmp as (
         select nvl(day.end_agent_code, last_day.end_agent_code)       as end_agent_code
              , nvl(day.end_agent_name, last_day.end_agent_name)       as end_agent_name
              , nvl(day.virt_code, last_day.virt_code)                 as virt_code
              , nvl(day.virt_name, last_day.virt_name)                 as virt_name
              , nvl(day.end_provider_id, last_day.end_provider_id)     as end_provider_id
              , nvl(day.end_provider_name, last_day.end_provider_name) as end_provider_name
              , nvl(day.district_code, last_day.district_code)         as district_code
              , nvl(day.district_desc, last_day.district_desc)         as district_desc
              , nvl(day.area_code, last_day.area_code)                 as area_code
              , nvl(day.area_name, last_day.area_name)                 as area_name
              , day.final_sign_network_code
              , last_day.final_sign_network_code                       as last_final_sign_network_code
         from day
                  full join last_day
                            on day.final_sign_network_code = last_day.final_sign_network_code
         where (day.final_sign_network_code is not null
             or last_day.final_sign_network_code is not null)
     ),
     result_a as (
         select virt_code                                               as virt_code
              , max(virt_name)                                          as virt_name
              , virt_code                                               as agent_code
              , max(virt_name)                                          as agent_name
              , sum(if(final_sign_network_code is not null, 1, 0))      as network_cnt
              , sum(if(last_final_sign_network_code is not null, 1, 0)) as before_networ_cnt
              , sum(
                 if(final_sign_network_code is not null and last_final_sign_network_code is null, 1,
                    0))                                                 as incr_networ_cnt
              , sum(
                 if(final_sign_network_code is null and last_final_sign_network_code is not null, 1,
                    0))                                                 as low_networ_cnt
              , '虚拟代理区'                                                 as agent_type
         from union_all_tmp
         where virt_code is not null
         group by virt_code
         union all
         select virt_code
              , max(virt_name)                                          as virt_name
              , end_agent_code                                          as agent_code
              , max(end_agent_name)                                     as agent_name
              , sum(if(final_sign_network_code is not null, 1, 0))      as network_cnt
              , sum(if(last_final_sign_network_code is not null, 1, 0)) as before_networ_cnt
              , sum(
                 if(final_sign_network_code is not null and last_final_sign_network_code is null, 1,
                    0))                                                 as incr_networ_cnt
              , sum(
                 if(final_sign_network_code is null and last_final_sign_network_code is not null, 1,
                    0))                                                 as low_networ_cnt
              , '代理区'                                                   as agent_type
         from union_all_tmp
         where end_agent_code is not null
         group by end_agent_code, virt_code
         union all
         select virt_code
              , max(virt_name)                                          as virt_name
              , end_provider_id                                         as agent_code
              , max(end_provider_name)                                  as agent_name
              , sum(if(final_sign_network_code is not null, 1, 0))      as network_cnt
              , sum(if(last_final_sign_network_code is not null, 1, 0)) as before_networ_cnt
              , sum(
                 if(final_sign_network_code is not null and last_final_sign_network_code is null, 1,
                    0))                                                 as incr_networ_cnt
              , sum(
                 if(final_sign_network_code is null and last_final_sign_network_code is not null, 1,
                    0))                                                 as low_networ_cnt
              , '省份'                                                    as agent_type
         from union_all_tmp
         where end_provider_id is not null
         group by end_provider_id, virt_code
         union all
         select virt_code
              , max(virt_name)                                          as virt_name
              , district_code                                             as agent_code
              , max(district_desc)                                      as agent_name
              , sum(if(final_sign_network_code is not null, 1, 0))      as network_cnt
              , sum(if(last_final_sign_network_code is not null, 1, 0)) as before_networ_cnt
              , sum(
                 if(final_sign_network_code is not null and last_final_sign_network_code is null, 1,
                    0))                                                 as incr_networ_cnt
              , sum(
                 if(final_sign_network_code is null and last_final_sign_network_code is not null, 1,
                    0))                                                 as low_networ_cnt
              , '片区'                                                    as agent_type
         from union_all_tmp
         where district_code is not null
         group by district_code, virt_code
         union all
         select virt_code
              , max(virt_name)                                          as virt_name
              , area_code                                               as agent_code
              , max(area_name)                                          as agent_name
              , sum(if(final_sign_network_code is not null, 1, 0))      as network_cnt
              , sum(if(last_final_sign_network_code is not null, 1, 0)) as before_networ_cnt
              , sum(
                 if(final_sign_network_code is not null and last_final_sign_network_code is null, 1,
                    0))                                                 as incr_networ_cnt
              , sum(
                 if(final_sign_network_code is null and last_final_sign_network_code is not null, 1,
                    0))                                                 as low_networ_cnt
              , '网管片区'                                                  as agent_type
         from union_all_tmp
         where area_code is not null
         group by area_code, virt_code
     )
insert
overwrite
table
jms_dm.dm_sqs_abnormal_network_cnt_dt
partition
(
dt
,
dy
)
select agent_code
     , agent_name
     , network_cnt
     , before_networ_cnt
     , incr_networ_cnt
     , ((network_cnt - before_networ_cnt) / before_networ_cnt) * 100 network_cnt_rt
     , low_networ_cnt
     , 0                               as                            target_value
     , network_cnt - 0                 as                            over_value
     , '爆仓'                            as                            data_type
     , '{{ execution_date | cst_ds }}' as                            date_time
     , agent_type
     , virt_code
     , virt_name
     , '{{ execution_date | cst_ds }}' as                            dt
     , 'baocang'                       as                            dy
from result_a
    distribute by dt
;

